package com.kai.mqtt.protocol;

import com.kai.mqtt.auth.service.IMqttUserService;
import com.kai.mqtt.constant.AttributeKeyConstant;
import com.kai.mqtt.message.DupPubRelMessage;
import com.kai.mqtt.message.DupPublishMessage;
import com.kai.mqtt.message.WillMessage;
import com.kai.mqtt.message.service.*;
import com.kai.mqtt.session.Session;
import com.kai.mqtt.session.service.ISessionService;
import io.netty.buffer.Unpooled;
import io.netty.channel.Channel;
import io.netty.handler.codec.DecoderResult;
import io.netty.handler.codec.mqtt.*;
import io.netty.handler.timeout.IdleStateHandler;
import jakarta.annotation.Resource;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import org.springframework.util.StringUtils;

import java.time.LocalDateTime;
import java.util.List;
import java.util.Objects;

/**
 * @author weijk
 * @since 2024/6/3
 */
@Slf4j
@Component
public class Connect extends AbstractProtocol {

    @Autowired
    private IMqttUserService mqttUserService;

    @Resource
    private ISessionService sessionService;

    @Resource
    private IWillMessageService willMessageService;

    @Resource
    private IDupPublishMessageService dupPublishMessageService;

    @Resource
    private IDupPubRelMessageService dupPubRelMessageService;

    @Resource
    private ISubscribeMessageService subscribeMessageService;

    @Resource
    private IMessageIdService messageIdService;

    @Override
    public void process(Channel channel, MqttMessage message) {
        MqttConnectMessage msg = (MqttConnectMessage) message;
        log.info("receive - connect, clientId: {}", msg.payload().clientIdentifier());
        DecoderResult decoderResult = msg.decoderResult();
        if (decoderResult.isFailure()) {
            Throwable cause = decoderResult.cause();
            if (cause instanceof MqttUnacceptableProtocolVersionException) {
                // 不支持的协议版本
                MqttConnAckMessage connAckMessage = (MqttConnAckMessage) MqttMessageFactory.newMessage(
                        new MqttFixedHeader(MqttMessageType.CONNACK, false, MqttQoS.AT_MOST_ONCE, false, 0),
                        new MqttConnAckVariableHeader(MqttConnectReturnCode.CONNECTION_REFUSED_UNACCEPTABLE_PROTOCOL_VERSION, false), null);
                log.info("send - connAck, message: 版本不支持");
                channel.writeAndFlush(connAckMessage);
            } else if (cause instanceof MqttIdentifierRejectedException) {
                // 不合格的 clintId
                MqttConnAckMessage connAckMessage = (MqttConnAckMessage) MqttMessageFactory.newMessage(
                        new MqttFixedHeader(MqttMessageType.CONNACK, false, MqttQoS.AT_MOST_ONCE, false, 0),
                        new MqttConnAckVariableHeader(MqttConnectReturnCode.CONNECTION_REFUSED_IDENTIFIER_REJECTED, false), null);
                log.info("send - connAck, message: clientId 不合格");
                channel.writeAndFlush(connAckMessage);
            }
            channel.close();
            return;
        }

        // clientId 为空或 null 的情况，这里要求客户端必须提供 clientId，不管 cleanSession 是否为1，此处没有参考标准协议实现
        if (!StringUtils.hasText(msg.payload().clientIdentifier())) {
            MqttConnAckMessage connAckMessage = (MqttConnAckMessage) MqttMessageFactory.newMessage(
                    new MqttFixedHeader(MqttMessageType.CONNACK, false, MqttQoS.AT_MOST_ONCE, false, 0),
                    new MqttConnAckVariableHeader(MqttConnectReturnCode.CONNECTION_REFUSED_IDENTIFIER_REJECTED, false), null);
            log.info("send - connAck, message: clientId 为 null");
            channel.writeAndFlush(connAckMessage);
            channel.close();
            return;
        }

        // 用户名和密码验证，这里要求客户端连接时必须提供用户名和密码，不管是否设置用户名标志和密码标志是否为1，此处没有参考标准协议实现
        byte[] passwordInBytes = msg.payload().passwordInBytes();
        String password = passwordInBytes == null ? null : new String(passwordInBytes);
        if (!mqttUserService.checkValid(msg.payload().userName(), password)) {
            MqttConnAckMessage connAckMessage = (MqttConnAckMessage) MqttMessageFactory.newMessage(
                    new MqttFixedHeader(MqttMessageType.CONNACK, false, MqttQoS.AT_MOST_ONCE, false, 0),
                    new MqttConnAckVariableHeader(MqttConnectReturnCode.CONNECTION_REFUSED_BAD_USER_NAME_OR_PASSWORD, false), null);
            log.info("send - connAck, message: 用户名或密码有误, clientId: {}, username: {}, password: {}", msg.payload().clientIdentifier(), msg.payload().userName(), password);
            channel.writeAndFlush(connAckMessage);
            channel.close();
            return;
        }

        // 如果会话中已储存这个新连接的 clientId, 就关闭之前该 clientId 的连接
        synchronized (msg.payload().clientIdentifier().intern()) {
            Session session = sessionService.get(msg.payload().clientIdentifier());
            if (Objects.nonNull(session) && session.getIsCleanSession()) {
                willMessageService.remove(msg.payload().clientIdentifier());
                subscribeMessageService.remove(msg.payload().clientIdentifier());
                messageIdService.remove(msg.payload().clientIdentifier());
                dupPublishMessageService.remove(msg.payload().clientIdentifier());
                dupPubRelMessageService.remove(msg.payload().clientIdentifier());
                sessionService.remove(msg.payload().clientIdentifier());
                Channel previous = session.getChannel();
                previous.attr(AttributeKeyConstant.CLIENT_ID).set(null);
                previous.close();
            }
        }

        boolean sessionPresent = Objects.nonNull(sessionService.get(msg.payload().clientIdentifier())) && !msg.variableHeader().isCleanSession();

        // 处理遗嘱消息
        if (msg.variableHeader().isWillFlag()) {
            WillMessage willMessage = new WillMessage()
                    .setTopic(msg.payload().willTopic())
                    .setQos(msg.variableHeader().willQos())
                    .setIsRetain(msg.variableHeader().isWillRetain())
                    .setMessageBytes(msg.payload().willMessageInBytes());
            willMessageService.put(msg.payload().clientIdentifier(), willMessage);
        }

        // 处理链接心跳包
        if (msg.variableHeader().keepAliveTimeSeconds() > 0) {
            if (channel.pipeline().names().contains("idle")) {
                channel.pipeline().remove("idle");
            }
            channel.pipeline().addFirst("idle", new IdleStateHandler(0, 0, Math.round(msg.variableHeader().keepAliveTimeSeconds() * 1.5f)));
        }

        // 保存会话信息，返回接收客户端连接
        Session session = new Session()
                .setChannel(channel)
                .setClientId(msg.payload().clientIdentifier())
                .setUsername(msg.payload().userName())
                .setIsCleanSession(msg.variableHeader().isCleanSession())
                .setKeepAliveTime(msg.variableHeader().keepAliveTimeSeconds())
                .setConnectTime(LocalDateTime.now());
        sessionService.put(session);
        channel.attr(AttributeKeyConstant.CLIENT_ID).set(msg.payload().clientIdentifier());
        MqttConnAckMessage connAckMessage = (MqttConnAckMessage) MqttMessageFactory.newMessage(
                new MqttFixedHeader(MqttMessageType.CONNACK, false, MqttQoS.AT_MOST_ONCE, false, 0),
                new MqttConnAckVariableHeader(MqttConnectReturnCode.CONNECTION_ACCEPTED, sessionPresent), null);
        log.info("send - connAck, message: success, clientId: {}", msg.payload().clientIdentifier());
        channel.writeAndFlush(connAckMessage);

        // 如果 sessionPresent 为 true，需要重发同一 clientId 存储的未完成的 qos1 和 qos2 的 dup 消息
        if (sessionPresent) {
            List<DupPublishMessage> dupPublishMessages = dupPublishMessageService.get(msg.payload().clientIdentifier());
            List<DupPubRelMessage> dupPubRelMessages = dupPubRelMessageService.get(msg.payload().clientIdentifier());
            dupPublishMessages.forEach(dupPublishMessage -> {
                MqttPublishMessage publishMessage = (MqttPublishMessage) MqttMessageFactory.newMessage(
                        new MqttFixedHeader(MqttMessageType.PUBLISH, true, MqttQoS.valueOf(dupPublishMessage.getQos()), false, 0),
                        new MqttPublishVariableHeader(dupPublishMessage.getTopic(), dupPublishMessage.getMessageId()), Unpooled.buffer().writeBytes(dupPublishMessage.getMessageBytes()));
                log.debug("send - publish, clientId: {}, topic: {}, qos: {}, messageId: {}, content: {}",
                        msg.payload().clientIdentifier(), dupPublishMessage.getTopic(), dupPublishMessage.getQos(), dupPublishMessage.getMessageId(), new String(dupPublishMessage.getMessageBytes()));
                channel.writeAndFlush(publishMessage);
            });
            dupPubRelMessages.forEach(dupPubRelMessage -> {
                MqttMessage pubRelMessage = MqttMessageFactory.newMessage(
                        new MqttFixedHeader(MqttMessageType.PUBREL, true, MqttQoS.AT_MOST_ONCE, false, 0),
                        MqttMessageIdVariableHeader.from(dupPubRelMessage.getMessageId()), null);
                log.debug("send - pubRel, clientId: {}, messageId: {}", msg.payload().clientIdentifier(), dupPubRelMessage.getMessageId());
                channel.writeAndFlush(pubRelMessage);
            });
        }
    }
}
